package com.apesay.hbase.connection;

import com.apesay.hbase.exception.HBaseAccessException;
import com.apesay.hbase.exception.HbaseSystemException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * ClassName:HBaseTemplate
 * Reason:	 
 * Date:     2018/3/26 16:26
 *
 * @author sid
 */
public class HBaseTemplate {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());


    private static final long DEFAULT_DESTORY_TIMEOUT = 2000;
    private static final int DEFAULT_MAX_THREADS_FOR_PARALLEL_SCANNER = 128;
    private static final int DEFAULT_MAX_THREADS_PER_PARALLEL_SCAN = 1;
    private ExecutorService executor;
    private boolean enableParallelScan = false;
    private int maxThreads = DEFAULT_MAX_THREADS_FOR_PARALLEL_SCANNER;
    private int maxThreadsPerParallelScan = DEFAULT_MAX_THREADS_PER_PARALLEL_SCAN;

    private Configuration configuration;

    private TableFactory tableFactory;

    private final AtomicBoolean isClose = new AtomicBoolean(false);
    
    public void initExecutorService(){
        HbaseThreadFactory parallelScannerThreadFactory = new HbaseThreadFactory("HBase-parallel-scanner");
        if (this.maxThreadsPerParallelScan <= 1) {
            this.enableParallelScan = false;
            this.executor = Executors.newSingleThreadExecutor(parallelScannerThreadFactory);
        } else {
            this.executor = ExecutorFactory.newFixedThreadPool(this.maxThreads, 1024, parallelScannerThreadFactory);
        }
    }


    public void put(TableName tableName, final byte[] rowName, final byte[] familyName, final byte[] qualifier, final byte[] value) {
        put(tableName, rowName, familyName, qualifier, null, value);
    }

    public void put(TableName tableName, final byte[] rowName, final byte[] familyName, final byte[] qualifier, final Long timestamp, final byte[] value) {
        assertAccessAvailable();
        execute(tableName, new TableCallback() {
            public Object doInTable(Table table) throws Throwable {
                Put put = createPut(rowName, familyName, timestamp, qualifier, value);
                table.put(put);
                return null;
            }
        });
    }

    public <T> T execute(TableName tableName, TableCallback<T> action) {
        assertAccessAvailable();

        Table table = getTable(tableName);

        try {
            T result = action.doInTable(table);
            return result;
        } catch (Throwable e) {
            if (e instanceof Error) {
                throw ((Error) e);
            }
            if (e instanceof RuntimeException) {
                throw ((RuntimeException) e);
            }
            throw new HbaseSystemException((Exception) e);
        } finally {
            releaseTable(table);
        }
    }
    
    private Put createPut(byte[] rowName, byte[] familyName, Long timestamp, byte[] qualifier, byte[] value) {
        Put put = new Put(rowName);
        if (familyName != null) {
            if (timestamp == null) {
                put.addColumn(familyName, qualifier, value);
            } else {
                put.addColumn(familyName, qualifier, timestamp, value);
            }
        }
        return put;
    }


    public void destroy() throws Exception {
//        long start = System.currentTimeMillis();

        if (isClose.compareAndSet(false, true)) {
            logger.info("HBaseTemplate2.destroy()");
            final ExecutorService executor = this.executor;
            if (executor != null) {
                executor.shutdown();
                try {
                    executor.awaitTermination(DEFAULT_DESTORY_TIMEOUT, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }

//            long remainingTime = Math.max(DEFAULT_DESTORY_TIMEOUT - (System.currentTimeMillis()-start), 100);
            
            //TODO 注意如果有异步操作，
//            awaitAsyncPutOpsCleared(remainingTime, 50);
        }
    }

    private void assertAccessAvailable() {
        if (isClose.get()) {
            throw new HBaseAccessException("Already closed.");
        }
    }

    private void releaseTable(Table table) {
        getTableFactory().releaseTable(table);
    }

    private Table getTable(TableName tableName) {
        return getTableFactory().getTable(tableName);
    }
    
    public Configuration getConfiguration() {
        return configuration;
    }

    public void setConfiguration(Configuration configuration) {
        this.configuration = configuration;
    }

    public TableFactory getTableFactory() {
        return tableFactory;
    }

    public void setTableFactory(TableFactory tableFactory) {
        this.tableFactory = tableFactory;
    }

}
